package simpleTaskDispatcher.model;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.junit.Test;
import org.omg.Dynamic.Parameter;

import simpleTaskDispatcher.model.WorkersPool;
import simpleTaskDispatcher.model.TaskQueue;
import simpleTaskDispatcher.model.enums.TaskPlacement;
import simpleTaskDispatcher.model.tasks.DelayedTasks;
import simpleTaskDispatcher.model.tasks.ITask;
import simpleTaskDispatcher.model.tasks.ParameterTasks;
import simpleTaskDispatcher.model.tasks.impl.AbstractTask;
import simpleTaskDispatcher.model.tasks.impl.ContextFreeTask;

import junit.framework.Assert;
import junit.framework.TestCase;

public class TaskQueueTest {

	@Test
	public void testOneTaskQueue1() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(22);

		TaskQueue taskQueue = new TaskQueue("t1", 1, 200);
		pool.addTaskQueue(taskQueue);

		while (pool.freeWorkersSize() != 0) {
			pool.doDispatching();
		}

		System.out.println("Main : After assignments");
		StringBuffer sb = new StringBuffer();

		long start = new Date().getTime();

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		// This barrier do nothing, as the 5 workers will finished at the same
		// moment
		taskQueue.barrier(false);

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		System.out.println("Main : After tasks creation");

		taskQueue.barrier(true);

		long duration = new Date().getTime() - start;
		TestCase.assertTrue(
				"Should take between 2000ms and 2999ms, but it take "
						+ duration + "ms.", duration > 2000 && duration < 2999);

		pool.closeWorkerPool();

		t.stopThread();

	}

	@Test
	public void testOneTaskQueue2() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(5);

		TaskQueue taskQueue = new TaskQueue("t", 1, 200);
		pool.addTaskQueue(taskQueue);

		while (pool.freeWorkersSize() != 0) {
			pool.doDispatching();
		}

		System.out.println("Main : After assignments");
		StringBuffer sb = new StringBuffer();

		long start = new Date().getTime();

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));

		// This barrier do nothing, as the 5 workers will finished at the same
		// moment
		taskQueue.barrier(false);
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		taskQueue.put(printAndWait(sb, "a", 1000));
		taskQueue.put(printAndWait(sb, "b", 1000));
		taskQueue.put(printAndWait(sb, "c", 1000));
		taskQueue.put(printAndWait(sb, "d", 1000));
		taskQueue.put(printAndWait(sb, "e", 1000));

		System.out.println("Main : After tasks creation");

		taskQueue.barrier(true);

		long duration = new Date().getTime() - start;
		TestCase.assertTrue(
				"Should take between 3000ms and 3999ms, but it take "
						+ duration + "ms.", duration > 3000 && duration < 3999);

		pool.closeWorkerPool();

		t.stopThread();
	}

	@Test
	public void testBarrierDeadLockFree1() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(100);

		TaskQueue taskQueue = new TaskQueue("t", 1, 200);
		pool.addTaskQueue(taskQueue);

		while (pool.freeWorkersSize() != 0) {
			pool.doDispatching();
		}

		// If it survive 1000 barriers, it is probably deadlock free
		for (int i = 0; i < 1000; i++) {
			taskQueue.barrier(false);

		}
		taskQueue.barrier(true);

		pool.closeWorkerPool();

		t.stopThread();
	}

	@Test
	public void testChangeMaxWorkers1() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(500);

		TaskQueue taskQueue = new TaskQueue("t", 1, 500);
		pool.addTaskQueue(taskQueue);

		while (pool.freeWorkersSize() != 0) {
			pool.doDispatching();
		}

		for (int releasedWorkers = 1; releasedWorkers < 500; releasedWorkers++) {
			System.out.println(releasedWorkers);
			int newMax = taskQueue.getMaximalWorkers() - 1;
			taskQueue.setMaximalWorkers(newMax,
					TaskPlacement.PLACE_AT_END_OF_QUEUE);

			taskQueue.barrier(true);
			TestCase.assertEquals(500 - releasedWorkers, newMax);
			TestCase.assertEquals(500 - releasedWorkers,
					taskQueue.getMaximalWorkers());

		}
		taskQueue.barrier(true);
		TestCase.assertEquals(1, taskQueue.plannedNbWorkers);
		taskQueue.close(true);
		TestCase.assertEquals(0, taskQueue.plannedNbWorkers);
		taskQueue.barrier(true);

		pool.closeWorkerPool();

		t.stopThread();
	}

	@Test
	public void testChangeMaxWorkers2() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(500);

		TaskQueue taskQueue = new TaskQueue("t", 1, 500);
		pool.addTaskQueue(taskQueue);

		while (pool.freeWorkersSize() != 0) {
			pool.doDispatching();
		}

		// If it survive 500 barriers, it is probably deadlock free
		for (int releasedWorkers = 1; releasedWorkers < 500; releasedWorkers++) {
			System.out.println(releasedWorkers);
			taskQueue.setMaximalWorkers(500 - releasedWorkers,
					TaskPlacement.PLACE_AT_END_OF_QUEUE);
			taskQueue.barrier(false);

		}
		taskQueue.barrier(true);
		TestCase.assertEquals(1, taskQueue.plannedNbWorkers);
		taskQueue.close(true);
		TestCase.assertEquals(0, taskQueue.plannedNbWorkers);
		taskQueue.barrier(true);

		pool.closeWorkerPool();

		t.stopThread();
	}

	@Test
	public void testChangeMaxWorkers3() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(500);

		TaskQueue taskQueue = new TaskQueue("t", 1, 500);
		pool.addTaskQueue(taskQueue);

		while (pool.freeWorkersSize() != 0) {
			pool.doDispatching();
		}

		// If it survive 500 barriers, it is probably deadlock free
		for (int releasedWorkers = 1; releasedWorkers < 500; releasedWorkers++) {
			System.out.println(releasedWorkers);
			int newMax = taskQueue.getMaximalWorkers() - 1;
			System.out.println(newMax);
			taskQueue.setMaximalWorkers(newMax,
					TaskPlacement.PLACE_AT_END_OF_QUEUE);
			taskQueue.barrier(true);

		}
		taskQueue.barrier(true);
		TestCase.assertEquals(1, taskQueue.plannedNbWorkers);
		taskQueue.close(true);
		TestCase.assertEquals(0, taskQueue.plannedNbWorkers);
		taskQueue.barrier(true);

		pool.closeWorkerPool();

		t.stopThread();
	}

	@Test
	public void testChangeMaxWorkers4() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(500);

		for (int i = 0; i < 1000; i++) {
			TaskQueue taskQueue = new TaskQueue("t", 1, 500);
			pool.addTaskQueue(taskQueue);

			List<TestTask> tasks = new ArrayList<TestTask>();
			for (int j = 0; j < 500; j++) {
				TestTask task = new TestTask();
				tasks.add(task);
			}

			while (pool.freeWorkersSize() != 0) {
				pool.doDispatching();
			}

			taskQueue.setMaximalWorkers(1, TaskPlacement.PLACE_AT_END_OF_QUEUE);
			for (int j = 0; j < 500; j++) {
				tasks.get(j).finished();
			}

			taskQueue.barrier(true);
			TestCase.assertEquals("On iteration " + i, 1,
					taskQueue.plannedNbWorkers);
			taskQueue.close(true);
			TestCase.assertEquals(0, taskQueue.plannedNbWorkers);
			taskQueue.barrier(true);
		}

		pool.closeWorkerPool();

		t.stopThread();
	}

	@Test
	public void testChangeMaxWorkers5() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(2);

		for (int i = 0; i < 1000; i++) {
			System.out.println("- - - - - - - -");
			TaskQueue taskQueue = new TaskQueue("t", 1, 2);
			pool.addTaskQueue(taskQueue);
			List<TestTask> tasks = new ArrayList<TestTask>();
			for (int j = 0; j < 2; j++) {
				TestTask task = new TestTask();
				taskQueue.put(task);
				tasks.add(task);
			}

			while (pool.freeWorkersSize() != 0) {
				pool.doDispatching();
			}

			System.out.println(taskQueue.getPlannedNbWorkers());

			System.out.println("before set");
			taskQueue.setMaximalWorkers(1, TaskPlacement.PLACE_AT_END_OF_QUEUE);
			for (int j = 0; j < 2; j++) {
				tasks.get(j).finished();
			}

			// Wait for the maximal worker event to be executed
			System.out.println("#" + taskQueue.plannedNbWorkers);
			while (pool.freeWorkersSize() != 1) {
				System.out.println("#" + taskQueue.plannedNbWorkers);
				pool.doDispatching();
			}

			TestCase.assertEquals("On iteration " + i, 1,
					taskQueue.plannedNbWorkers);
			taskQueue.barrier(true);
			TestCase.assertEquals("On iteration " + i, 1,
					taskQueue.plannedNbWorkers);
			taskQueue.close(true);
			TestCase.assertEquals(0, taskQueue.plannedNbWorkers);
			taskQueue.barrier(true);
		}

		pool.closeWorkerPool();

		t.stopThread();
	}

	/**
	 * 
	 * @throws InterruptedException
	 */
	@Test
	public void testWithALotOfTaskQueues() throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		WorkersPool pool = new WorkersPool();
		pool.createWorkers(1);

		for (int i = 0; i < 1000; i++) {
			TaskQueue taskQueue = new TaskQueue("t" + i, 1, 1);
			pool.addTaskQueue(taskQueue);
			List<TestTask> tasks = new ArrayList<TestTask>();
			while (pool.freeWorkersSize() != 0) {
				pool.doDispatching();
			}
			for (int j = 0; j < 1; j++) {
				TestTask task = new TestTask();
				tasks.add(task);
			}
			while (taskQueue.plannedNbWorkers != 1) {
				pool.doDispatching();
				Thread.sleep(100);
			}
			for (int j = 0; j < 1; j++) {
				tasks.get(j).finished();
			}
			TestCase.assertEquals("On iteration " + i, 1,
					taskQueue.plannedNbWorkers);
			taskQueue.barrier(true);
			TestCase.assertEquals("On iteration " + i, 1,
					taskQueue.plannedNbWorkers);
			taskQueue.close(true);
			TestCase.assertEquals(0, taskQueue.plannedNbWorkers);
			TestCase.assertEquals(1, pool.freeWorkersSize());
			// taskQueue.barrier(true);
		}

		pool.closeWorkerPool();

		t.stopThread();
	}

	@Test
	/**
	 * Can the pool be closed sucessfully? (all threads and task queues closed)
	 * @throws InterruptedException
	 */
	public void testPoolClosingWithoutTaskQueueClosing()
			throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		for (int i = 0; i < 500; i++) {
			System.out.println("Testing with pool " + i);
			WorkersPool pool = new WorkersPool();
			pool.createWorkers(500);
			pool.closeWorkerPool();
		}

		t.stopThread();
	}

	/**
	 * Can the pool be closed sucessfully? (all threads and task queues closed)
	 * The call is made when all task queues are closed
	 * 
	 * @throws InterruptedException
	 */
	@Test
	public void testPoolClosingWithTaskQueueBlocking()
			throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		for (int i = 0; i < 500; i++) {
			System.out.println("Testing with pool " + i);
			WorkersPool pool = new WorkersPool();
			pool.createWorkers(500);
			TaskQueue queue = new TaskQueue("f" + i);
			pool.addTaskQueue(queue);
			queue.close(true);
			pool.closeWorkerPool();
		}

		t.stopThread();
	}

	@Test
	public void testPoolClosingWithoutTaskQueueBlocking()
			throws InterruptedException {

		WTFThread t = new WTFThread();
		t.start();

		for (int i = 0; i < 500; i++) {
			System.out.println("Testing with pool " + i);
			WorkersPool pool = new WorkersPool();
			pool.createWorkers(500);
			TaskQueue queue = new TaskQueue("f" + i);
			pool.addTaskQueue(queue);
			queue.close(false);
			pool.closeWorkerPool();
		}

		t.stopThread();
	}

	@Test
	public void testProgression() throws InterruptedException {

		WorkersPool pool = new WorkersPool();

		pool.createWorkers(2);

		TaskQueue queue1 = new TaskQueue("q1");
		TaskQueue queue2 = new TaskQueue("q2");
		pool.addTaskQueue(queue1);
		pool.addTaskQueue(queue2);

		BlockingQueue<String> eventsLogs = new LinkedBlockingQueue<String>();

		TestTask task1_1 = new TestTask("1_1", eventsLogs).setWorkValue(10);
		queue1.put(task1_1);

		TestTask task1_2 = new TestTask("1_2", eventsLogs).setWorkValue(10);
		queue1.put(task1_2);

		TestTask task1_3 = new TestTask("1_3", eventsLogs).setWorkValue(0);
		queue1.put(task1_3);

		TestTask task1_4 = new TestTask("1_4", eventsLogs).setWorkValue(10);
		queue1.put(task1_4);

		TestTask task2_1 = new TestTask("2_1", eventsLogs).setWorkValue(10);
		queue2.put(task2_1);

		TestTask task2_2 = new TestTask("2_2", eventsLogs).setWorkValue(10);
		queue2.put(task2_2);

		TestTask task2_3 = new TestTask("2_3", eventsLogs).setWorkValue(0);
		queue2.put(task2_3);

		TestTask task2_4 = new TestTask("2_4", eventsLogs).setWorkValue(10);
		queue2.put(task2_4);

		Assert.assertEquals(30.0, queue1.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue1.getTotalWorkValue());
		Assert.assertEquals(30.0, queue1.getUnrealisedWork());
		Assert.assertEquals(0.0, queue1.getRealisedWork());
		Assert.assertEquals(0.0f, queue1.getProgression());

		Assert.assertEquals(30.0, queue2.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue2.getTotalWorkValue());
		Assert.assertEquals(30.0, queue2.getUnrealisedWork());
		Assert.assertEquals(0.0, queue2.getRealisedWork());
		Assert.assertEquals(0.0f, queue2.getProgression());

		Assert.assertEquals(60.0, pool.getUnrealisedWorkValue());
		Assert.assertEquals(0.0f, pool.getProgression());

		assertInLogs(eventsLogs, "start of 1_1", "start of 2_1");

		task1_1.finished();
		task2_1.finished();
		assertInLogs(eventsLogs, "end of 1_1", "end of 2_1", "start of 1_2",
				"start of 2_2");

		Assert.assertEquals(30.0, queue1.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue1.getTotalWorkValue());
		Assert.assertEquals(20.0, queue1.getUnrealisedWork());
		Assert.assertEquals(10.0, queue1.getRealisedWork());
		Assert.assertEquals(0.3333, queue1.getProgression(), 0.0005);

		Assert.assertEquals(30.0, queue2.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue2.getTotalWorkValue());
		Assert.assertEquals(20.0, queue2.getUnrealisedWork());
		Assert.assertEquals(10.0, queue2.getRealisedWork());
		Assert.assertEquals(0.3333, queue2.getProgression(), 0.0005);

		Assert.assertEquals(40.0, pool.getUnrealisedWorkValue());
		Assert.assertEquals(0.3333, pool.getProgression(), 0.0005);

		task1_2.finished();
		assertInLogs(eventsLogs, "end of 1_2", "start of 1_3");

		Assert.assertEquals(30.0, queue1.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue1.getTotalWorkValue());
		Assert.assertEquals(10.0, queue1.getUnrealisedWork());
		Assert.assertEquals(20.0, queue1.getRealisedWork());
		Assert.assertEquals(0.6666, queue1.getProgression(), 0.0005);

		Assert.assertEquals(30.0, queue2.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue2.getTotalWorkValue());
		Assert.assertEquals(20.0, queue2.getUnrealisedWork());
		Assert.assertEquals(10.0, queue2.getRealisedWork());
		Assert.assertEquals(0.3333, queue2.getProgression(), 0.0005);

		Assert.assertEquals(30.0, pool.getUnrealisedWorkValue());
		Assert.assertEquals(0.5, pool.getProgression(), 0.0005);

		task1_3.finished();
		assertInLogs(eventsLogs, "end of 1_3", "start of 1_4");

		Assert.assertEquals(30.0, queue1.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue1.getTotalWorkValue());
		Assert.assertEquals(10.0, queue1.getUnrealisedWork());
		Assert.assertEquals(20.0, queue1.getRealisedWork());
		Assert.assertEquals(0.6666, queue1.getProgression(), 0.0005);

		Assert.assertEquals(30.0, queue2.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue2.getTotalWorkValue());
		Assert.assertEquals(20.0, queue2.getUnrealisedWork());
		Assert.assertEquals(10.0, queue2.getRealisedWork());
		Assert.assertEquals(0.3333, queue2.getProgression(), 0.0005);

		Assert.assertEquals(30.0, pool.getUnrealisedWorkValue());
		Assert.assertEquals(0.5, pool.getProgression(), 0.0005);

		task1_4.finished();
		assertInLogs(eventsLogs, "end of 1_4");

		Assert.assertEquals(30.0, queue1.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue1.getTotalWorkValue());
		Assert.assertEquals(0.0, queue1.getUnrealisedWork());
		Assert.assertEquals(30.0, queue1.getRealisedWork());
		Assert.assertEquals(1.0, queue1.getProgression(), 0.0005);

		Assert.assertEquals(30.0, queue2.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue2.getTotalWorkValue());
		Assert.assertEquals(20.0, queue2.getUnrealisedWork());
		Assert.assertEquals(10.0, queue2.getRealisedWork());
		Assert.assertEquals(0.3333, queue2.getProgression(), 0.0005);

		Assert.assertEquals(20.0, pool.getUnrealisedWorkValue());
		Assert.assertEquals(0.6666, pool.getProgression(), 0.0005);

		queue1.setEstimatedWorkValue(60.0);

		Assert.assertEquals(30.0, queue1.getCurrentTotalWork());
		Assert.assertEquals(60.0, queue1.getTotalWorkValue());
		Assert.assertEquals(30.0, queue1.getUnrealisedWork());
		Assert.assertEquals(30.0, queue1.getRealisedWork());
		Assert.assertEquals(0.5, queue1.getProgression(), 0.0005);

		Assert.assertEquals(30.0, queue2.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue2.getTotalWorkValue());
		Assert.assertEquals(20.0, queue2.getUnrealisedWork());
		Assert.assertEquals(10.0, queue2.getRealisedWork());
		Assert.assertEquals(0.3333, queue2.getProgression(), 0.0005);

		Assert.assertEquals(50.0, pool.getUnrealisedWorkValue());
		Assert.assertEquals(40.0, pool.getRealisedWorkValue());
		Assert.assertEquals(90.0, pool.getTotalWorkValue());
		Assert.assertEquals(40.0 / 90, pool.getProgression(), 0.0005);

		queue1.close(true);
		Assert.assertEquals(30.0, queue1.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue1.getTotalWorkValue());
		Assert.assertEquals(0.0, queue1.getUnrealisedWork());
		Assert.assertEquals(30.0, queue1.getRealisedWork());
		Assert.assertEquals(1.0, queue1.getProgression(), 0.0005);

		Assert.assertEquals(30.0, queue2.getCurrentTotalWork());
		Assert.assertEquals(30.0, queue2.getTotalWorkValue());
		Assert.assertEquals(20.0, queue2.getUnrealisedWork());
		Assert.assertEquals(10.0, queue2.getRealisedWork());
		Assert.assertEquals(0.3333, queue2.getProgression(), 0.0005);

		// Now, there is only one task queue
		Assert.assertEquals(20.0, pool.getUnrealisedWorkValue());
		Assert.assertEquals(10.0, pool.getRealisedWorkValue());
		Assert.assertEquals(30.0, pool.getTotalWorkValue());
		Assert.assertEquals(0.3333, pool.getProgression(), 0.0005);

		assertInLogs(eventsLogs, "start of 2_3");

		task2_2.finished();
		task2_3.finished();
		task2_4.finished();

		queue1.close(false);
		queue2.close(false);

		pool.closeWorkerPool();

	}

	private void assertInLogs(BlockingQueue<String> eventsLogs,
			String... logEvents) throws InterruptedException {
		while (eventsLogs.size() < logEvents.length) {
			Thread.sleep(100);
		}
		Thread.sleep(100);

		List<String> expectedEvents = new ArrayList<String>();
		for (String logEvent : logEvents) {
			expectedEvents.add(logEvent);
		}

		List<String> events = new ArrayList<String>();
		String event = null;
		while ((event = eventsLogs.poll()) != null) {
			events.add(event);
		}

		Collections.sort(expectedEvents);
		Collections.sort(events);

		TestCase.assertEquals(expectedEvents, events);

	}

	/**
	 * This task wait the specified wait time, then append a text to a sting
	 * buffer
	 * 
	 * @return
	 */
	public ITask printAndWait(final StringBuffer sb, final String text,
			final long wait) {
		return new ContextFreeTask() {

			@Override
			public void execute() throws InterruptedException {
				sb.append(text);
				Thread.sleep(wait);
			}

		}.setWorkValue(0);
	}

	@Test
	public void testSharedParameterValues() throws InterruptedException {

		WorkersPool p = new WorkersPool();
		p.createWorkers(2);
		TaskQueue taskQueue1 = new TaskQueue("TaskQueue1", 1, 3);
		p.addTaskQueue(taskQueue1);
		TaskQueue taskQueue2 = new TaskQueue("TaskQueue2", 1, 3);
		p.addTaskQueue(taskQueue2);

		taskQueue1.put(ParameterTasks.incrementInteger("a"));
		taskQueue1.put(ParameterTasks.incrementInteger("a"));
		taskQueue1.put(ParameterTasks.incrementInteger("a"));
		taskQueue1.put(ParameterTasks.incrementInteger("a"));
		taskQueue1.put(ParameterTasks.incrementInteger("b"));
		taskQueue1.put(ParameterTasks.incrementInteger("b"));
		taskQueue1.put(ParameterTasks.incrementInteger("b"));
		taskQueue1.put(ParameterTasks.incrementInteger("b"));

		taskQueue1.barrier(true);

		Thread.sleep(1000);

		Assert.assertEquals(4, taskQueue1.getSharedParameter("a"));
		Assert.assertEquals(4, taskQueue1.getSharedParameter("b"));
		Assert.assertNull(taskQueue2.getSharedParameter("a"));
		Assert.assertNull(taskQueue2.getSharedParameter("b"));

	}

	private static ITask contextWithAttribute(final String contextID,
			final String name, final Object value) {
		return new AbstractTask() {

			@Override
			public void loadContext(Worker w) {
				w.set(name, value);
			}

			@Override
			public String getContextID() {
				return contextID;
			}

			@Override
			public void execute() throws InterruptedException {
				Object contextParamValue = getContextParameter(name);
				setSharedParameter(name, contextParamValue);
			}
		};
	}

	private void waitExecuted(ITask t) throws InterruptedException {
		while (!t.isExecuted())
			Thread.sleep(100);
	}

	@Test
	public void testPutAfter() throws InterruptedException {
		WorkersPool p = new WorkersPool();
		p.createWorkers(1);

		TaskQueue taskQueue1 = new TaskQueue("TaskQueue1", 1, 3);
		p.addTaskQueue(taskQueue1);

		ITask t1 = ParameterTasks.incrementInteger("n");
		ITask d1 = ParameterTasks.decrementInteger("n");
		taskQueue1.put(t1);
		taskQueue1.put(new DelayedTasks(d1, t1,
				TaskPlacement.PLACE_AT_END_OF_QUEUE));

		waitExecuted(d1);

		// d2 executed before t3 before of a priority task placement
		TestCase.assertEquals(0, taskQueue1.getSharedParameter("n"));
		ITask d2 = ParameterTasks.decrementInteger("n");
		ITask t2 = ParameterTasks.incrementInteger("n");
		ITask t3 = ParameterTasks.multiplyInteger("n", 2);

		taskQueue1.put(new DelayedTasks(d2, t2,
				TaskPlacement.PLACE_AT_START_OF_QUEUE));
		taskQueue1.put(t2);
		taskQueue1.put(t3);

		TestCase.assertEquals(0, taskQueue1.getSharedParameter("n"));
		waitExecuted(d2);
		TestCase.assertEquals(0, taskQueue1.getSharedParameter("n"));

		ITask t4 = ParameterTasks.incrementInteger("n");
		ITask d3 = ParameterTasks.decrementInteger("n");
		taskQueue1.put(t4);
		waitExecuted(t4);
		TestCase.assertEquals(1, taskQueue1.getSharedParameter("n"));
		taskQueue1.put(new DelayedTasks(d3, t4,
				TaskPlacement.PLACE_AT_START_OF_QUEUE));

		waitExecuted(d3);
		TestCase.assertEquals(0, taskQueue1.getSharedParameter("n"));

		ITask t5 = ParameterTasks.incrementInteger("n");
		ITask t6 = ParameterTasks.incrementInteger("n");
		ITask d4 = ParameterTasks.multiplyInteger("n", 3);

		taskQueue1.put(new DelayedTasks(d4, Arrays
				.asList(new ITask[] { t5, t6 }),
				TaskPlacement.PLACE_AT_END_OF_QUEUE));
		taskQueue1.put(t5);
		taskQueue1.put(t6);
		taskQueue1.close(true);
		waitExecuted(d4);
		TestCase.assertEquals(6, taskQueue1.getSharedParameter("n"));



		taskQueue1 = new TaskQueue("TaskQueue1", 1, 3);
		p.addTaskQueue(taskQueue1);
		
		// d5 executed after t8 before of a normal task placement
		ITask d5 = ParameterTasks.decrementInteger("n");
		ITask t7 = ParameterTasks.incrementInteger("n");
		ITask t8 = ParameterTasks.multiplyFloat("n", 2);

		taskQueue1.put(new DelayedTasks(d5, t7,
				TaskPlacement.PLACE_AT_END_OF_QUEUE));
		taskQueue1.put(t7);
		taskQueue1.put(t8);
		waitExecuted(d5);
		TestCase.assertEquals(1, taskQueue1.getSharedParameter("n"));

		
		
		//Multiple levels of dependencies
		ITask d6 = ParameterTasks.incrementInteger("n");
		ITask d7 = ParameterTasks.multiplyInteger("n",3);
		ITask d8 = ParameterTasks.decrementInteger("n",8);
		ITask t9 = ParameterTasks.incrementInteger("n");
		

		taskQueue1.put(t9);
		taskQueue1.put(new DelayedTasks(d6, t9,
				TaskPlacement.PLACE_AT_END_OF_QUEUE));
		taskQueue1.put(new DelayedTasks(Arrays.asList(d7, d8), d6,
				TaskPlacement.PLACE_AT_END_OF_QUEUE));
		
		waitExecuted(d7);
		TestCase.assertEquals(1, taskQueue1.getSharedParameter("n"));
	}

	@Test
	public void testContexts() throws InterruptedException {

		WorkersPool p = new WorkersPool();
		p.createWorkers(1);
		Worker worker = (Worker) p.allWorkers.get(0);
		TaskQueue taskQueue1 = new TaskQueue("TaskQueue1", 1, 3);
		p.addTaskQueue(taskQueue1);
		TaskQueue taskQueue2 = new TaskQueue("TaskQueue2", 1, 3);
		p.addTaskQueue(taskQueue2);

		taskQueue1.put(contextWithAttribute("a", "a", 1));
		taskQueue1.barrier(true);
		Assert.assertEquals(1, taskQueue1.getSharedParameter("a"));
		Assert.assertEquals(1, worker.get("a"));

		taskQueue1.put(contextWithAttribute("b", "b", 1));
		taskQueue1.barrier(true);
		Assert.assertEquals(1, taskQueue1.getSharedParameter("a"));
		Assert.assertEquals(1, taskQueue1.getSharedParameter("b"));
		Assert.assertNull(worker.get("a"));
		Assert.assertEquals(1, worker.get("b"));

		taskQueue1.put(contextWithAttribute(null, "b", 9));
		taskQueue1.barrier(true);
		Assert.assertEquals(1, taskQueue1.getSharedParameter("a"));
		Assert.assertEquals(1, taskQueue1.getSharedParameter("b"));
		Assert.assertNull(worker.get("a"));
		Assert.assertEquals(1, worker.get("b"));

		taskQueue1.put(contextWithAttribute("a", "a", 2));
		taskQueue1.barrier(true);
		Assert.assertEquals(2, taskQueue1.getSharedParameter("a"));
		Assert.assertEquals(1, taskQueue1.getSharedParameter("b"));
		Assert.assertEquals(2, worker.get("a"));
		Assert.assertNull(worker.get("b"));

		taskQueue1.barrier(true);

		Assert.assertEquals(2, taskQueue1.getSharedParameter("a"));
		Assert.assertEquals(1, taskQueue1.getSharedParameter("b"));

		Assert.assertNull(taskQueue2.getSharedParameter("a"));
		Assert.assertNull(taskQueue2.getSharedParameter("b"));

	}

	@Test
	public void main() {
		System.out.println(ParameterTasks.decrementDouble("n").getUniqueID());
		System.out.println(ParameterTasks.decrementInteger("n").getUniqueID());
		System.out.println(ParameterTasks.decrementInteger("n").getUniqueID());
	}
}
